package com.hrt.bigdata.mr.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * windows提交mr到集群方式运行
 */
public class DriverOnYarn {

    //bin/hadoop command [genericOptions] [commandOptions]
    //    hadoop jar  ooxx.jar  ooxx   -D  ooxx=ooxx  inpath  outpath
    //  args :   2类参数  genericOptions   commandOptions
    //  人你有复杂度：  自己分析 args数组
    //
    public static void main(String[] args) throws Exception {
        // true表示加载resources目录下的配置文件
        Configuration conf = new Configuration(true);
        //工具类帮我们把-D 等等的属性直接set到conf，会留下commandOptions
        GenericOptionsParser parser = new GenericOptionsParser(conf, args);
        String[] othargs = parser.getRemainingArgs();

        //让框架知道是windows异构平台运行
        conf.set("mapreduce.app-submission.cross-platform","true");
//        conf.set("mapreduce.framework.name","local");
//        System.out.println(conf.get("mapreduce.framework.name"));
        Job job = Job.getInstance(conf);

//        FileInputFormat.setMinInputSplitSize(job,2222);
//        job.setInputFormatClass(ooxx.class);
//        提前用package打好jar
        job.setJar("F:\\P8\\workspace\\hadoop-p8\\hadoop-mr\\target\\hadoop-mr-1.0-SNAPSHOT.jar");
        //必须必须写的
        job.setJarByClass(DriverOnYarn.class);
        job.setJobName("hrt");

        Path infile = new Path(othargs[0]);
//        Path infile = new Path("/wordcount/test.txt");
        TextInputFormat.addInputPath(job, infile);
        // 如果输出目录存在则删除
//        Path outfile = new Path("/wordcount_result/");
        Path outfile = new Path(othargs[1]);
        if (outfile.getFileSystem(conf).exists(outfile)) outfile.getFileSystem(conf).delete(outfile, true);
        TextOutputFormat.setOutputPath(job, outfile);

        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setReducerClass(MyReducer.class);

//        job.setNumReduceTasks(2);
        // Submit the job, then poll for progress until the job is complete
        job.waitForCompletion(true);

    }

}
